package com.sly.fuxi.kafka.producer;

import com.sly.fuxi.kafka.constant.KafkaTopicConstant;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author SLY
 * @date 2025/2/17
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    @Resource
    private KafkaProducer<String, String> specificKafkaTemplate;

    @RequestMapping("/sendMessage")
    public Object sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
        return "success";
    }

    @RequestMapping("/sendPartitionMessage")
    public Object sendPartitionMessage(String message) {
        ProducerRecord<String, String> record = new ProducerRecord<>(KafkaTopicConstant.SPECIFIC_TOPIC, "key", message);
        specificKafkaTemplate.send(record);
        return "success";
    }
}
